package com.sh.data.engine.domain.dataquality.job;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.enumDefinition.DataQualityCategoryEnum;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.DateUtil;
import com.sh.data.engine.domain.common.service.CommonService;
import com.sh.data.engine.domain.dataasset.model.domain.TableInfoBizDomain;
import com.sh.data.engine.domain.dataasset.model.param.QueryTableParam;
import com.sh.data.engine.domain.dataasset.service.TableService;
import com.sh.data.engine.domain.dataquality.service.*;
import com.sh.data.engine.domain.shims.db.BaseDbManager;
import com.sh.data.engine.domain.shims.db.model.FieldInfoDomain;
import com.sh.data.engine.domain.shims.db.model.TableInfoDomain;
import com.sh.data.engine.domain.workspace.index.service.ProjectService;
import com.sh.data.engine.job.core.handler.annotation.XxlJob;
import com.sh.data.engine.repository.dao.base.EngineBaseModel;
import com.sh.data.engine.repository.dao.dataquality.entity.*;
import com.sh.data.engine.repository.dao.workspace.index.entity.ProjectEntity;
import com.sh.data.engine.repository.dao.workspace.manager.entity.StorageEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author tinglan.ys
 * @description:
 * @date 2022/3/2 上午11:44
 */
@Component
@Slf4j
public class DataQualityStatisticsTask {

    @Autowired
    private DataQualityRuleService dataQualityRuleService;

    @Autowired
    private DataQualityStatisticsExceptionService dataQualityStatisticsExceptionService;

    @Autowired
    private DataQualityJobRuleService dataQualityJobRuleService;

    @Autowired
    private DataQualityJobRuleFieldService dataQualityJobRuleFieldService;

    @Autowired
    private DataQualityStatisticsMonitorService dataQualityStatisticsMonitorService;

    @Autowired
    private DataQualityJobService dataQualityJobService;

    @Autowired
    private DataQualityStatisticsTableExceptionService dataQualityStatisticsTableExceptionService;

    @Autowired
    private ProjectService projectService;

    @Autowired
    private TableService tableService;

    @Autowired
    private CommonService commonService;

    private final List<Integer> templateCategories = Lists.newArrayList(0, 1, 2, 3);

    /**
     * @description : 分类统计数据质量异常数 每10分钟执行一次 cron = "0 0/10 * * * ?"
     * @author : tinglan.ys
     * @date : 2022/3/9
     */
    @XxlJob(value = "statisticsException")
    public void statisticsException() {
        log.info("开始进行分类统计数据质量异常数statisticsException");
        String statisticsDay = DateFormatUtils.format(new Date(), "yyyy-MM-dd");

        // 对所有空间的进行数据质量统计
        List<ProjectEntity> projects = projectService.findAll();

        if (CollectionUtils.isNotEmpty(projects)) {
            for (ProjectEntity project : projects) {
                try {

                    // 查询该空间下的所有规则
                    List<DataQualityRuleEntity> rules =
                            dataQualityRuleService.list(
                                    new LambdaQueryWrapper<DataQualityRuleEntity>()
                                            .eq(DataQualityRuleEntity::getProjectId, project.getId()));

                    if (null == rules) {
                        rules = new ArrayList<>();
                    }

                    // 统计自定义规则的异常数
                    List<Long> diyRuleIds =
                            rules.stream()
                                    .filter(r -> r.getRuleDiy() == 1)
                                    .map(EngineBaseModel::getId)
                                    .collect(Collectors.toList());
                    int diyExceptionCount = 0;
                    if (CollectionUtils.isNotEmpty(diyRuleIds)) {
                        diyExceptionCount = statisticsException(diyRuleIds);
                    }
                    /** 分类 0-完整性 1-准确性 2-唯一性 3-有效性 diy规则为准确性 */
                    this.saveStatisticsException(
                            project.getId(),
                            DataQualityCategoryEnum.ZQX.getId(),
                            statisticsDay,
                            diyExceptionCount);

                    // 统计模版规则的异常数
                    for (Integer templateCategory : templateCategories) {
                        List<Long> ruleIds =
                                rules.stream()
                                        .filter(r -> Objects.equals(r.getTemplateCategory(), templateCategory))
                                        .map(EngineBaseModel::getId)
                                        .collect(Collectors.toList());

                        int exceptionCount = 0;

                        if (CollectionUtils.isNotEmpty(ruleIds)) {
                            // 当日数据质量异常数
                            exceptionCount = statisticsException(ruleIds);
                        }

                        this.saveStatisticsException(
                                project.getId(), templateCategory, statisticsDay, exceptionCount);
                    }
                } catch (Exception e) {
                    log.error("分类统计数据质量异常数任务执行时有异常，项目id:{}", project.getId(), e);
                }
            }
        }
    }

    private void saveStatisticsException(
            Long projectId, Integer category, String statisticsDay, Integer exceptionCount) {

        LambdaQueryWrapper<DataQualityStatisticsExceptionEntity> queryWrapper =
                new LambdaQueryWrapper<DataQualityStatisticsExceptionEntity>()
                        .eq(DataQualityStatisticsExceptionEntity::getProjectId, projectId)
                        .eq(DataQualityStatisticsExceptionEntity::getStatisticsDay, statisticsDay);
        if (null == category) {
            queryWrapper.isNull(DataQualityStatisticsExceptionEntity::getCategory);
        } else {
            queryWrapper.eq(DataQualityStatisticsExceptionEntity::getCategory, category);
        }

        DataQualityStatisticsExceptionEntity exception =
                dataQualityStatisticsExceptionService.getOne(queryWrapper);

        if (exception == null) {
            DataQualityStatisticsExceptionEntity statisticsException =
                    new DataQualityStatisticsExceptionEntity();
            statisticsException.setStatisticsDay(statisticsDay);
            statisticsException.setProjectId(projectId);
            statisticsException.setCategory(category);
            statisticsException.setExceptionCount(exceptionCount);
            dataQualityStatisticsExceptionService.save(statisticsException);
        } else {
            exception.setExceptionCount(exceptionCount);
            dataQualityStatisticsExceptionService.updateById(exception);
        }
    }

    /**
     * @description : 统计表数、字段数信息 cron = "0 2/10 * * * ?"
     * @author : tinglan.ys
     * @date : 2022/4/13
     */
    @XxlJob(value = "statisticsMonitor")
    public void statisticsMonitor() {
        log.info("开始进行统计表数、字段数信息statisticsMonitor");
        List<ProjectEntity> projects = projectService.findAll();

        if (CollectionUtils.isNotEmpty(projects)) {
            for (ProjectEntity project : projects) {
                try {

                    List<DataQualityRuleEntity> rules =
                            dataQualityRuleService.list(
                                    new LambdaQueryWrapper<DataQualityRuleEntity>()
                                            .eq(DataQualityRuleEntity::getProjectId, project.getId()));

                    // 首先先查询出所有的表 统计
                    // 带描述的总条数，表描述算一条，字段描述算一条，包括分区字段
                    // 不带描述的总条数

                    BaseDbManager dbManager = commonService.getStorageDbManagerByProjctId(project.getId());
                    StorageEntity storageEntity = commonService.getStorage(project.getId());

                    if (dbManager == null || storageEntity == null) {
                        continue;
                    }
                    String dbName = storageEntity.getRmdbs().getDbName();
                    if (StringUtils.isEmpty(dbName)) {
                        throw new BusinessException("dbName为空");
                    }

                    QueryTableParam param = new QueryTableParam();
                    List<TableInfoBizDomain> tableList = tableService.getTableList(project.getId(), param);

                    int totalTableDescCount = 0;
                    int totalTableNoDescCount = 0;

                    for (TableInfoDomain tableInfoDomain : tableList) {
                        // 表级的注释
                        if (tableInfoDomain.getTableComment() == null) {
                            totalTableNoDescCount++;
                        } else {
                            totalTableDescCount++;
                        }
                        // 字段注释
                        String tableName = removeTableSchema(tableInfoDomain.getTableName());
                        List<FieldInfoDomain> fieldList = dbManager.getFieldList(dbName, tableName);
                        for (FieldInfoDomain fieldInfoDomain : fieldList) {
                            if (StringUtils.isBlank(fieldInfoDomain.getComment())) {
                                totalTableNoDescCount++;
                            } else {
                                totalTableDescCount++;
                            }
                        }
                    }

                    List<Long> allTblIds = new ArrayList<>();
                    List<String> allTblNames = new ArrayList<>();

                    if (CollectionUtils.isNotEmpty(tableList)) {
                        allTblIds =
                                tableList.stream().map(TableInfoDomain::getTblId).collect(Collectors.toList());
                        allTblNames =
                                tableList.stream().map(TableInfoDomain::getTableName).collect(Collectors.toList());
                    }

                    Set<Long> tblIds = new HashSet<>();
                    Set<String> fields = new HashSet<>();

                    if (CollectionUtils.isNotEmpty(rules)) {
                        for (DataQualityRuleEntity dataqualityRule : rules) {
                            Long tblId = dataqualityRule.getTblId();
                            if (!allTblIds.contains(tblId)) {
                                continue;
                            }

                            //              String tblName =
                            //                      tableList.stream()
                            //                              .filter(t -> Objects.equals(t.getTblId(), tblId))
                            //                              .findFirst()
                            //                              .get()
                            //                              .getTableName();
                            //              if (!allTblNames.contains(tblName.toLowerCase())) {
                            //                continue;
                            //              }

                            tblIds.add(tblId);

                            String fieldNames = dataqualityRule.getFieldNames();
                            if (StringUtils.isNotBlank(fieldNames)) {
                                String[] split = fieldNames.split(",");
                                for (String s : split) {
                                    if (StringUtils.isNotBlank(s)) {
                                        String field = String.format("%s.%s", dataqualityRule.getTblId(), s);
                                        fields.add(field);
                                    }
                                }
                            }
                        }
                    }

                    DataQualityStatisticsMonitorEntity statisticsMonitor =
                            dataQualityStatisticsMonitorService.getOne(
                                    new LambdaQueryWrapper<DataQualityStatisticsMonitorEntity>()
                                            .eq(DataQualityStatisticsMonitorEntity::getProjectId, project.getId())
                                            .eq(
                                                    DataQualityStatisticsMonitorEntity::getStatisticsDay,
                                                    DateFormatUtils.format(new Date(), "yyyy-MM-dd")));

                    if (null == statisticsMonitor) {
                        DataQualityStatisticsMonitorEntity monitor = new DataQualityStatisticsMonitorEntity();
                        monitor.setProjectId(project.getId());
                        monitor.setStatisticsDay(DateFormatUtils.format(new Date(), "yyyy-MM-dd"));
                        monitor.setTableCount(tblIds.size());
                        monitor.setFieldCount(fields.size());
                        monitor.setTotalTableCount(allTblNames.size());
                        monitor.setTotalTableDescCount(totalTableDescCount);
                        monitor.setTotalTableNoDescCount(totalTableNoDescCount);
                        dataQualityStatisticsMonitorService.save(monitor);
                    } else {
                        statisticsMonitor.setTableCount(tblIds.size());
                        statisticsMonitor.setFieldCount(fields.size());
                        statisticsMonitor.setTotalTableCount(allTblNames.size());
                        statisticsMonitor.setTotalTableDescCount(totalTableDescCount);
                        statisticsMonitor.setTotalTableNoDescCount(totalTableNoDescCount);
                        dataQualityStatisticsMonitorService.updateById(statisticsMonitor);
                    }
                } catch (Exception e) {
                    log.error("统计表数、字段数信息任务执行时有异常，项目id:{}", project.getId(), e);
                }
            }
        }
    }

    /**
     * @description : 统计表的异常数 cron = "0 4/10 * * * ?"
     * @author : tinglan.ys
     * @date : 2022/4/13
     */
    @XxlJob("statisticsTableException")
    public void statisticsTableException() {
        log.info("开始进行统计表的异常数statisticsTableException");
        List<ProjectEntity> projects = projectService.findAll();

        String statisticsDate = DateFormatUtils.format(new Date(), "yyyy-MM-dd");

        if (CollectionUtils.isNotEmpty(projects)) {
            for (ProjectEntity project : projects) {

                try {

                    List<DataQualityRuleEntity> rules =
                            dataQualityRuleService.list(
                                    new LambdaQueryWrapper<DataQualityRuleEntity>()
                                            .eq(DataQualityRuleEntity::getProjectId, project.getId()));

                    if (CollectionUtils.isEmpty(rules)) {
                        continue;
                    }
                    // 首先先查询出所有的表
                    QueryTableParam param = new QueryTableParam();
                    List<TableInfoBizDomain> tableList = tableService.getTableList(project.getId(), param);

                    if (CollectionUtils.isEmpty(tableList)) {
                        dataQualityStatisticsTableExceptionService.remove(
                                new LambdaQueryWrapper<DataQualityStatisticsTableExceptionEntity>()
                                        .eq(DataQualityStatisticsTableExceptionEntity::getProjectId, project.getId())
                                        .eq(
                                                DataQualityStatisticsTableExceptionEntity::getStatisticsDay,
                                                statisticsDate));

                        continue;
                    }

                    List<Long> allTblIds =
                            tableList.stream().map(t -> t.getTblId()).collect(Collectors.toList());

                    List<Long> tblIds =
                            rules.stream().map(r -> r.getTblId()).distinct().collect(Collectors.toList());

                    for (Long tblId : tblIds) {

                        // 如果表不存在，则删除该表的统计记录
                        if (!allTblIds.contains(tblId)) {

                            dataQualityStatisticsTableExceptionService.remove(
                                    new LambdaQueryWrapper<DataQualityStatisticsTableExceptionEntity>()
                                            .eq(DataQualityStatisticsTableExceptionEntity::getProjectId, project.getId())
                                            .eq(
                                                    DataQualityStatisticsTableExceptionEntity::getStatisticsDay,
                                                    statisticsDate)
                                            .eq(DataQualityStatisticsTableExceptionEntity::getTblId, tblId));

                            continue;
                        }

                        int tableExceptionCount = countTableException(project.getId(), tblId);

                        DataQualityStatisticsTableExceptionEntity tableException =
                                dataQualityStatisticsTableExceptionService.getOne(
                                        new LambdaQueryWrapper<DataQualityStatisticsTableExceptionEntity>()
                                                .eq(
                                                        DataQualityStatisticsTableExceptionEntity::getProjectId,
                                                        project.getId())
                                                .eq(DataQualityStatisticsTableExceptionEntity::getTblId, tblId)
                                                .eq(
                                                        DataQualityStatisticsTableExceptionEntity::getStatisticsDay,
                                                        statisticsDate));

                        String tblName =
                                tableList.stream()
                                        .filter(t -> Objects.equals(t.getTblId(), tblId))
                                        .findFirst()
                                        .get()
                                        .getTableName();

                        if (null == tableException) {
                            tableException = new DataQualityStatisticsTableExceptionEntity();
                            tableException.setProjectId(project.getId());
                            tableException.setTblId(tblId);
                            tableException.setStatisticsDay(statisticsDate);
                            tableException.setTblName(tblName);
                            tableException.setExceptionCount(tableExceptionCount);
                            dataQualityStatisticsTableExceptionService.save(tableException);
                        } else {
                            tableException.setTblName(tblName);
                            tableException.setExceptionCount(tableExceptionCount);
                            dataQualityStatisticsTableExceptionService.updateById(tableException);
                        }
                    }

                    // 删除表不存在的统计结果
                    dataQualityStatisticsTableExceptionService.remove(
                            new LambdaQueryWrapper<DataQualityStatisticsTableExceptionEntity>()
                                    .eq(DataQualityStatisticsTableExceptionEntity::getProjectId, project.getId())
                                    .eq(DataQualityStatisticsTableExceptionEntity::getStatisticsDay, statisticsDate)
                                    .notIn(DataQualityStatisticsTableExceptionEntity::getTblId, allTblIds));

                } catch (Exception e) {
                    log.error("统计表的异常数任务执行时有异常,项目id:{}", project.getId(), e);
                }
            }
        }
    }

    /**
     * @description : 统计异常数
     * @author : tinglan.ys
     * @date : 2022/4/15
     */
    public int statisticsException(List<Long> diyRuleIds) {

    /*SELECT
    count( * )
    FROM
    dataquality_job_rule_field jrf
    LEFT JOIN dataquality_job_rule jr ON jrf.dataquality_job_rule_id = jr.id
    WHERE
    TO_DAYS( jrf.update_time ) = TO_DAYS( now( ) )
    <if test="ruleIds != null and ruleIds.size() > 0">
        and jr.rule_id in
        <foreach collection="ruleIds" index="index" item="ruleId" open="(" separator="," close=")">
            #{ruleId}
        </foreach>
    </if>*/

        List<DataQualityJobRuleEntity> dataQualityJobRuleEntityList =
                dataQualityJobRuleService.list(
                        new LambdaQueryWrapper<DataQualityJobRuleEntity>()
                                .in(DataQualityJobRuleEntity::getRuleId, diyRuleIds));

        if (dataQualityJobRuleEntityList.isEmpty()) {
            return 0;
        }


        List<Long> ruleIds =
                dataQualityJobRuleEntityList.stream().map(dj -> dj.getId()).collect(Collectors.toList());

        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
        Date date = new Date();
        String day = format.format(date) + " 00:00:00";
        Date today = DateUtil.parse(day, DateUtil.DATE_PATTERN.yyyyMMddHHmmss);

        // ge表示大于等于
        Long count =
                dataQualityJobRuleFieldService.count(
                        new LambdaQueryWrapper<DataQualityJobRuleFieldEntity>()
                                .in(DataQualityJobRuleFieldEntity::getDataqualityJobRuleId, ruleIds)
                                .in(DataQualityJobRuleFieldEntity::getStatus, Lists.newArrayList(2, 3))
                                .ge(DataQualityJobRuleFieldEntity::getUpdateTime, today));

        String s = count.toString();
        return Integer.parseInt(s);
    }

    /**
     * @description : 统计表的异常数
     * @author : tinglan.ys
     * @date : 2022/3/31
     */
    private int countTableException(Long projectId, Long tblId) {

    /* select
     count(1)
    FROM
        dataquality_job_rule_field jrf
        LEFT JOIN dataquality_job_rule jr ON jrf.dataquality_job_rule_id = jr.id
        LEFT JOIN dataquality_job j ON jr.dataquality_job_id = j.id
    WHERE
        jrf.`status` IN ( 2, 3 )
        AND j.tbl_id = #{tblId}
        AND j.project_id = #{projectId}*/
        // 查询数据质量任务

        List<DataQualityJobEntity> dataQualityJobEntities =
                dataQualityJobService.list(
                        new LambdaQueryWrapper<DataQualityJobEntity>()
                                .eq(DataQualityJobEntity::getProjectId, projectId)
                                .eq(DataQualityJobEntity::getTblId, tblId));

        // 查询规则执行记录
        List<Long> allJobIds =
                dataQualityJobEntities.stream().map(EngineBaseModel::getId).collect(Collectors.toList());

        if (allJobIds.isEmpty()) {
            return 0;
        }
        List<DataQualityJobRuleEntity> dataQualityJobRuleEntities =
                dataQualityJobRuleService.list(
                        new LambdaQueryWrapper<DataQualityJobRuleEntity>()
                                .in(DataQualityJobRuleEntity::getDataqualityJobId, allJobIds));

        // 查询每个字段的执行记录
        List<Long> allJobRuleIds =
                dataQualityJobRuleEntities.stream().map(EngineBaseModel::getId).collect(Collectors.toList());

        List<DataQualityJobRuleFieldEntity> dataQualityJobRuleFieldEntities =
                dataQualityJobRuleFieldService.list(
                        new LambdaQueryWrapper<DataQualityJobRuleFieldEntity>()
                                .in(DataQualityJobRuleFieldEntity::getDataqualityJobRuleId, allJobIds)
                                .in(DataQualityJobRuleFieldEntity::getStatus, Lists.newArrayList(2, 3)));
        return dataQualityJobRuleFieldEntities.size();
    }

    private static String removeTableSchema(String fullTableName) {
        // 找到最后一个点的位置
        int lastIndex = fullTableName.lastIndexOf('.');

        // 如果没有点号，返回整个字符串；否则返回最后一个点号后面的内容
        return (lastIndex == -1) ? fullTableName : fullTableName.substring(lastIndex + 1);
    }
}
